package yxy.banana.socket.controller.client;

import yxy.apple.protobuf.Constants;
import yxy.apple.protobuf.Message.HeartbeatRequest;
import yxy.apple.protobuf.Message.HeartbeatResponse;
import yxy.apple.protobuf.Message.Parcal;
import yxy.apple.util.IDGenerator;
import yxy.apple.util.scheduler.Schedulable;
import yxy.apple.util.scheduler.ScheduledFuture;
import yxy.apple.util.scheduler.SingleThreadScheduler;
import yxy.banana.socket.NetworkElement;
import yxy.banana.socket.controlled.ControlledHandler;
import yxy.banana.socket.controlled.Controller;
import yxy.banana.socket.receiver.Observer;
import yxy.banana.socket.receiver.Receivable;

public abstract class ClientHeartbeatController<H extends ControlledHandler> extends Controller<H> {

	private static final SingleThreadScheduler<Schedulable> scheduler = new SingleThreadScheduler<>();

	private class Pulse implements Schedulable {
		private int interval = 10000;
		private int timeout = 5000;

		@Override
		public void trigger() {
			if (ClientHeartbeatController.this.beating) {
				ClientHeartbeatController.this.beat(this.interval, this.timeout);
			}
		}

		@Override
		public void cancelled() {
		}
	}

	private ScheduledFuture feture = null;
	private int silencedTimes = 0;
	private boolean beating = false;

	public boolean beating() {
		return this.beating;
	}

	/**
	 * 心跳起跳，一般应在actived中启动
	 * 
	 * @param interval 心跳周期
	 * @param timeout  等待心跳回应超时
	 */
	public void pace(int interval, int timeout) {
		if (this.beating) {
			return;
		}
		this.beating = true;
		this.beat(interval, timeout);
	}

	private void beat(int interval, int timeout) {
		HeartbeatRequest.Builder builder = HeartbeatRequest.newBuilder();
		builder.setTimestamp(System.currentTimeMillis());
		HeartbeatRequest request = builder.build();
		Receivable receiver = new Receivable() {
			@Override
			public void receive(Parcal parcal) {
				if (parcal.getProto() == Constants.ControlProto.HeartbeatResponse) {
					try {
						HeartbeatResponse response = HeartbeatResponse.parseFrom(parcal.getData());
						long delta = System.currentTimeMillis() - response.getTimestamp();
						ClientHeartbeatController.this.announced(delta);
					} catch (Exception e) {
						ClientHeartbeatController.this.silenced(++ClientHeartbeatController.this.silencedTimes);
						ClientHeartbeatController.this.caught(e);
					}
				} else {
					ClientHeartbeatController.this.silenced(++ClientHeartbeatController.this.silencedTimes);
					ClientHeartbeatController.this.drop(parcal, DroppedReason.UnknownProto);
				}
			}

			@Override
			public void exipred(String tag) {
				ClientHeartbeatController.this.silenced(++ClientHeartbeatController.this.silencedTimes);
			}
		};
		try {
			String tag = IDGenerator.nextUUID();
			Observer.shared.observe(tag, timeout, receiver);
			long s = NetworkElement.value(this.handler().system(), this.handler().cluster(), this.handler().channel());
			long d = 0;
			this.handler().send(tag, s, d, Constants.ControlProto.HeartbeatRequest, request);
		} catch (Exception e) {
			ClientHeartbeatController.this.silenced(++this.silencedTimes);
			this.caught(e);
		}
		// 注册下一次脉搏
		Pulse pulse = new Pulse();
		pulse.interval = interval;
		pulse.timeout = timeout;
		this.feture = ClientHeartbeatController.scheduler.schedule(pulse, interval);
	}

	/**
	 * 心脏停跳，一般在inactived时调用
	 */
	public void arrest() {
		if (this.feture != null) {
			this.feture.cancel();
			this.feture = null;
		}
		this.beating = false;
	}

	/**
	 * 收到心跳回应时回调
	 * 
	 * @param handler
	 * @param delta   心跳请求与回应的时间差
	 */
	protected abstract void announced(long delta);

	/**
	 * 未正常收到心跳回应时回调
	 * 
	 * @param handler
	 * @param times   已连续n次未收到，任何一次正常收到都会重置为0
	 */
	protected abstract void silenced(int times);

}
